package com.afdor.rws.event.channel.aliyun.mq;

import com.afdor.rws.event.Event;
import com.afdor.rws.event.channel.aliyun.mq.utils.AliyunMQUtils;
import com.afdor.rws.event.channel.client.consumer.AbstractEventConsumer;
import com.afdor.rws.event.config.EventConfigUtils;
import com.aliyun.mns.client.CloudAccount;
import com.aliyun.mns.client.CloudQueue;
import com.aliyun.mns.client.MNSClient;
import com.aliyun.mns.model.Message;
import lombok.EqualsAndHashCode;
import lombok.Getter;
import lombok.ToString;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.math.NumberUtils;

import java.util.ArrayList;
import java.util.List;

/**
 * 基于阿里云MQ的抽象事件消费者实现
 *
 * @author 悭梵
 * @date Created in 2018-08-16 14:59
 */
@Slf4j
@ToString
@EqualsAndHashCode
public abstract class AbstractAliyunEventConsumer extends AbstractEventConsumer {

    @Getter
    private CloudAccount cloudAccount;

    private CloudQueue cloudQueue;
    @Getter
    private volatile int waitSeconds = -1;
    @Getter
    private volatile int batchSize = 50;


    @Override
    public void start() {
        cloudAccount = new CloudAccount(EventConfigUtils.getProperty(AliyunMQChannelProvider.ALIYUN_MQ_SECRET_ID), EventConfigUtils.getProperty(AliyunMQChannelProvider.ALIYUN_MQ_SECRET_KEY), EventConfigUtils.getProperty(AliyunMQChannelProvider.ALIYUN_MQ_ENDPOINT));
        CloudQueue queue = getCloudQueue();
        if (!queue.isQueueExist()) {
            try {
                queue.create();
            } catch (Exception e) {
                log.error("创建队列异常", e);
            }
        }
        String secondStr = EventConfigUtils.getProperty(EventConfigUtils.EVENT_CONSUMER_GET_MESSAGE_TIMEOUT, "-1");
        if (!NumberUtils.isCreatable(secondStr)) {
            secondStr = "-1";
        }
        this.waitSeconds = NumberUtils.createInteger(secondStr);
        String batchSizeStr = EventConfigUtils.getProperty(EventConfigUtils.EVENT_CONSUMER_GET_MESSAGE_BATCH_SIZE, "8");
        if (!NumberUtils.isCreatable(batchSizeStr)) {
            batchSizeStr = "8";
        }
        this.batchSize = NumberUtils.createInteger(batchSizeStr);
        super.start();
    }


    @Override
    protected boolean consume() {
        if (batchSize > 1) {
            return batchHandlerMessage();
        } else {
            return handlerMessage();
        }
    }

    /**
     * 批量处理消息事件，采用batchPopMessage同步处理实现
     *
     * @return
     */
    protected boolean batchHandlerMessage() {
        CloudQueue cloudQueue = getCloudQueue();
        List<Message> messageList = waitSeconds > 0 ? cloudQueue.batchPopMessage(batchSize, waitSeconds) : cloudQueue.batchPopMessage(batchSize);
        if (messageList == null || messageList.isEmpty()) {
            return Boolean.FALSE;
        } else {
            List<String> receiptHandleList = new ArrayList<>(messageList.size());
            messageList.parallelStream().forEach(message -> {
                receiptHandleList.add(handlerMessage(message));
            });
            cloudQueue.batchDeleteMessage(receiptHandleList);
            return Boolean.TRUE;
        }
    }

    /**
     * 处理单条消息，采用popMessage同步处理实现
     *
     * @return
     */
    protected boolean handlerMessage() {
        CloudQueue cloudQueue = getCloudQueue();
        Message message = waitSeconds > 0 ? cloudQueue.popMessage(waitSeconds) : cloudQueue.popMessage();
        if (message == null) {
            return Boolean.FALSE;
        }
        cloudQueue.deleteMessage(handlerMessage(message));
        return Boolean.TRUE;
    }


    /**
     * 实际处理事件消息
     *
     * @param message
     * @return
     */
    protected String handlerMessage(Message message) {
        Event event = getSerialization().unserialize(message.getMessageBodyAsRawString());
        if (event == null) {
            return message.getReceiptHandle();
        }
        try {
            if (log.isInfoEnabled()) {
                log.info("<===== 收到消息：" + event);
            }
            getEventHandler().handler(event);
        } catch (Exception e) {
            if (getEventRetryStrategy() != null) {
                getEventRetryStrategy().retry(event);
            }
            log.error("事件处理异常", e);
        }
        return message.getReceiptHandle();
    }


    protected CloudQueue getCloudQueue() {
        if (cloudQueue == null) {
            synchronized (this) {
                if (cloudQueue == null) {
                    MNSClient mnsClient = cloudAccount.getMNSClient();
                    cloudQueue = mnsClient.getQueueRef(AliyunMQUtils.getQueueName(getEventSourceName()));
                }
            }
        }
        return cloudQueue;
    }


}
